/**************************************************************************
 *
 * Copyright (C) 2007 University of California, Los Angeles.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 *
 *************************************************************************/

/**
 * 
 */
package edu.ucla.cs.typecast.test;

import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintStream;
import java.net.InetAddress;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

import edu.ucla.cs.typecast.net.EndPoint;
import edu.ucla.cs.typecast.net.NetUtil;
import edu.ucla.cs.typecast.rmi.Future;
import edu.ucla.cs.typecast.rmi.FutureListener;
import edu.ucla.cs.typecast.rmi.TypeCastClient;
import edu.ucla.cs.typecast.rmi.TypeCastServer;

/**
 *
 * @date Nov 25, 2007
 */
public class PingFuture implements PingAck, FutureListener<PingAcknowledgement> {
	private static final Object logLock = new Object(); 
	
	private long delay; 
	private String sessionID; 
	private long duration; 
	private int rate; 
	private long interval; 
	private EndPoint serviceEndPoint; 
	private int payloadSize; 
	private OutputStream ostream; 
	private PrintStream latencyOutStream; 
	
	private Ping pingClient; 
	//private TypeCastServer ackServer; 

	private long seqno = 1; 
	private byte[] payload; 
	private long totalSend = 0; 
	private long totalAcked = 0; 
	private double totalLatency = 0; 
	private double aveLatency = 0; 	
	private Timer timer = new Timer(); 
	private boolean init = false; 
	private Class targetType; 
	private String targetDisplayType; 
	private long startEffectiveCount = -1; 
	private long totalEffectiveCount = 0; 
	private double totalEffectiveLatency = 0; 
	private double aveEffectiveLatency = 0; 
		
	private Map<Long, Date> sendTimeMap = Collections.synchronizedMap(new HashMap<Long, Date>(1024));
	TimerTask sendTask; 
	
	public PingFuture(Class targetType, EndPoint endPoint, String sessionID, long delay, long duration, int rate, int payloadSize, OutputStream output, OutputStream latencyOutputStream) { 
		this.targetType = targetType; 
		this.targetDisplayType = getDisplayName(targetType); 
		this.delay = delay; 
		this.serviceEndPoint = endPoint; 
		this.sessionID = sessionID; 
		this.duration = duration; 
		this.rate = rate; 
		this.payloadSize = payloadSize; 
		this.interval = 1000 / rate; 		
		this.payload = new byte[payloadSize]; 
		this.ostream = output; 
		this.startEffectiveCount = rate * 60; // count after 1 minute; 
		if (latencyOutputStream != null) { 
			this.latencyOutStream = new PrintStream(latencyOutputStream); 
		}
		
		start(); 
	}
	
	private void start() { 
		try { 
			TimerTask cancelTask = new TimerTask() { 
				public void run() { 
					try { 
						sendTask.cancel(); 
					} catch(Throwable t) { 
						t.printStackTrace(); 
					}
				}
			}; 
			timer.schedule(cancelTask, delay + duration); 			
			
			
			// Schedule timer to send packets
			sendTask = new TimerTask() { 
				public void run() { 
					try { 
						send(); 
					} catch(Throwable t) { 
						t.printStackTrace(); 
					}
				}
			}; 
			
			timer.scheduleAtFixedRate(sendTask, delay, interval); 
			
			TimerTask stopTask = new TimerTask() { 
				public void run() { 
					try { 
						timer.cancel(); 
						shutdown(); 
					} catch(Throwable t) { 
						t.printStackTrace(); 
					}
					try { 
						PrintStream logStream = new PrintStream(ostream); 
						saveStatistics(logStream); 
						if(latencyOutStream != null) { 
							latencyOutStream.flush(); 
						}
					} catch(Throwable t) { 
						t.printStackTrace(); 
					}
				}
			};
			// Stop everthing 5 minutes after the sending task is done. 
			timer.schedule(stopTask, delay + duration + 1 * 60 * 1000); 
			
		} catch(Throwable t) { 
			t.printStackTrace(); 
		}
	}
	
	/* (non-Javadoc)
	 * @see edu.ucla.cs.typecast.test.PingAck#pong(java.net.InetAddress, java.lang.String, java.lang.String)
	 */
	public synchronized void pingAck(InetAddress source, InetAddress server, String sessionID, long seqno) {
		// TODO Auto-generated method stub
		if (sessionID.equals(this.sessionID)) { 
			Date sendTime = this.sendTimeMap.get(new Long(seqno)); 
			if (sendTime == null) { 
				System.out.println("Error: missing send timestamp for session " + sessionID + " seqno=" + seqno); 
				return; 
			}
			totalAcked++; 
			Date receiveTime = new Date(); 
			this.totalLatency += receiveTime.getTime() - sendTime.getTime(); 
			this.aveLatency = totalLatency / totalAcked; 		
			if (seqno > this.startEffectiveCount) { 
				this.totalEffectiveCount++; 
				this.totalEffectiveLatency += receiveTime.getTime() - sendTime.getTime();
				this.aveEffectiveLatency = this.totalEffectiveLatency / this.totalEffectiveCount; 
			}			
			if(this.latencyOutStream != null) { 
				latencyOutStream.print(targetDisplayType); 
				latencyOutStream.print(" "); 
				latencyOutStream.print(this.serviceEndPoint); 
				latencyOutStream.print(" "); 
				latencyOutStream.print(NetUtil.getLocalHost().getHostAddress()); 
				latencyOutStream.print(" "); 
				latencyOutStream.print(this.sessionID); 
				latencyOutStream.print(" "); 
				latencyOutStream.print(seqno); 
				latencyOutStream.print(" "); 				
				latencyOutStream.print(receiveTime.getTime() - sendTime.getTime()); 
				latencyOutStream.print(""); 
				latencyOutStream.println(); 
			}
		} else { 
			System.out.println("Error received invalid sessionID " + sessionID); 
		}
	}
	
	private synchronized void send() { 
		if (!init) { 
			init(); 
		}
		sendTimeMap.put(new Long(seqno), new Date()); 
		Future<PingAcknowledgement> future = pingClient.ping3(NetUtil.getLocalHost(),  sessionID, seqno++, payload); 
		totalSend++; 
		future.addFutureListener(this);
		future.freeze(20000, -1); 

	}
	
	private void init() { 
		try { 
			/*
			ackServer = createTypeCastServerForAck(); 
			ackServer.export(PingAck.class, this); 
			*/
			TypeCastClient tclient = new TypeCastClient(serviceEndPoint); 
			pingClient = (Ping)tclient.getTypeCastInvocationHandler(targetType); 
			init = true; 
		} catch(Throwable t) { 
			t.printStackTrace(); 
		}
	}
	
	private void shutdown() { 
		//this.ackServer.unexport(this); 		
	}
	protected EndPoint getTargetEndPoint() { 
		return this.serviceEndPoint;
	}
	/*
	protected TypeCastServer createTypeCastServerForAck() throws IOException { 
		return  new TypeCastServer(null); 		
	}
	*/
	
	public void saveStatistics(PrintStream out) { 		
		try { 
			synchronized(logLock) { 
				Date now = new Date(); 				
				out.print(targetDisplayType); 
				out.print(" "); 
				out.print(this.serviceEndPoint); 
				out.print(" "); 
				out.print(NetUtil.getLocalHost().getHostAddress()); 
				out.print(" "); 
				out.print(this.sessionID); 
				out.print(" "); 
				out.print(this.duration/1000); 
				out.print(" "); 
				out.print(this.payloadSize); 
				out.print(" "); 
				out.print(this.rate); 
				out.print(" "); 
				out.print(this.totalSend); 
				out.print(" "); 
				out.print(this.totalAcked); 
				out.print(" "); 
				out.print(((double)totalAcked) / totalSend); 
				out.print(" "); 
				out.print(this.aveLatency);
				out.print(" "); 
				out.print(this.aveEffectiveLatency); 
				out.print(" "); 
				out.print(now); 
				out.println(""); 
				out.flush(); 
			} 
		} catch(Throwable t) { 
			t.printStackTrace(); 
		}
	}
	
	public String getDisplayName(Class type) { 
		String className = type.getName(); 
		return className.substring("edu.ucla.cs.typecast.test.Ping".length()); 
	}
	
	public void newValueArrived(Future<PingAcknowledgement> future, PingAcknowledgement value) { 
		pingAck(value.getSource(), null, value.getSessionID(), value.getSeqno());
	}
	
	public void freezed(Future<PingAcknowledgement> future) { 
		
	}
}
